
#include <cstring>
#include <stdexcept>

#include <string>
#include <sstream>
#include <fstream>

#include <boost/lexical_cast.hpp>
#include <boost/filesystem.hpp>

#include "wbsinker_error.hpp"
#include "fifo_storage.hpp"

#include "wbsinker.hpp"


#define	ROUND_COUNT_MAX		50000


///////////////////////////////////////////////////////////////////////
//

WbSinker::WbSinker( const char* sink_config_filename, 
					const char* log_config_filename )
		:	_storage_p( NULL )
{
	if ( NULL != log_config_filename ) {
		QPub::qLogConfig( log_config_filename );
	}

	_storage_p = new FifoStorage( sink_config_filename );
	_storage_p->open();

	_round_count = 0;
	_round_sum_bytes = 0;
}

WbSinker::~WbSinker()
{
	if ( NULL != _storage_p ) {
		_storage_p->close();
		delete _storage_p;
	}
}


///////////////////////////////////////////////////////////////////////
//


static	float
getTimeDuration( ptime time_begin, ptime time_end )
{
    time_period tp( time_begin, time_end );
    time_duration   td = tp.length();
    long    time_last_ms = td.total_milliseconds();
    float   time_last_s = time_last_ms / 1000.0f;

    return time_last_s;
}


int		
WbSinker::handle( const std::string& data )
{
	if ( data.size() < 1 ) {
		return	0;
	}
	if ( _round_count == 0 ) {
		_round_time_start = microsec_clock::local_time();
	}
	++_round_count;
	_round_sum_bytes += data.size();

	if ( _round_count > ROUND_COUNT_MAX ) {
		ptime   _round_time_stop = microsec_clock::local_time();
		float   time_last_s = getTimeDuration( _round_time_start, _round_time_stop );
		float   tps = _round_count / time_last_s;
		float   bw = _round_sum_bytes / time_last_s / 1048576.0;
		INFO(CAT_SINKER, "  SINK %u records last %.3f seconds, tps=%.2f, bw=%.2f MB/s", _round_count - 1, time_last_s, tps, bw);

		_round_count = 0;
		_round_sum_bytes = 0;
	}


	const char*	type_id_p = data.c_str();

	const char* tab_p = strchr( type_id_p, '\t' );
	if ( NULL == tab_p ) {
		// BAD FORMAT : no type_id
		ERROR( CAT_SINKER, "BAD FORMAT : no type id in data '%s'", data.c_str() ); 
		return	-1;
	}
	size_t	type_id_size = tab_p - type_id_p;
	if ( type_id_size > 32 ) {
		// BAD FORMAT : invalid type_id
		ERROR( CAT_SINKER, "BAD FORMAT : invalid type id in data '%s'", data.c_str() ); 
		return	-2;
	}

	char	buf[ 64 ] = {0};
	strncpy( buf, type_id_p, type_id_size );

	int	type_id = 0;
	try {
		type_id = boost::lexical_cast< int >( buf );
	}
	catch( std::exception& e ) {
		ERROR( CAT_SINKER, "fail to lexical_cast '%s' -- %s", buf, e.what() ); 
		return	-3;
	}

	const char*	data_p = tab_p + 1;
	size_t		data_size = data.size() - type_id_size - 1;
	_storage_p->write( type_id, data_p, data_size );

	return	0;
}


///////////////////////////////////////////////////////////////////////
//

extern	"C"
boque::trasit::Handler*
create( const char* p1, const char* p2)
{
	if ( (strlen(p1) < 1) ) {
		std::ostringstream	msg;
		msg << "handler-p1 should be sink-config-filename";
		throw std::runtime_error( msg.str() );
	}

	if ( (strlen(p2) < 1) ) {
		std::ostringstream	msg;
		msg << "handler-p2 should be log-config-filename";
		throw std::runtime_error( msg.str() );
	}

	/*
	boost::filesystem::path	path1( p1 );
	boost::filesystem::path	p_path1 = path1.parent_path();
	if ( !boost::filesystem::exists( p_path1 ) ) {
        try {
            boost::filesystem::create_directories( p_path1 );
        }   
        catch ( std::exception& e ) { 
            std::ostringstream  msg;
            msg << "fail to create directories " << p_path1.string() << " -- " << e.what();
            throw   std::runtime_error( msg.str() );
        }   
	}
	*/

	WbSinker*	handler_p = new WbSinker( p1, p2 );

	return	handler_p;
}


extern	"C"
void
destroy( boque::trasit::Handler* p ) 
{
	if ( NULL != p ) {
		delete p;
	}
}

